package com.sunway.resource2es.config;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.sunway.exception.built.BusinessException;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.config.ConnectionConfig;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.nio.charset.StandardCharsets;

/**
 * @author tan
 */
@Configuration
public class ElasticClientConfig {

    private static final Logger LOGGER = LoggerFactory.getLogger(ElasticClientConfig.class);

    @Value("${spring.elasticsearch.uris}")
    private String hosts;

    @Value("${spring.elasticsearch.username}")
    private String userName;

    @Value("${spring.elasticsearch.password}")
    private String passWord;

    /**
     * 连接超时/毫秒 （-1 默认）
     *
     * @see RequestConfig#getConnectTimeout()
     */
    @Value("${spring.elasticsearch.connection-timeout:5000}")
    private Integer connectTimeout;

    /**
     * 套接超时/毫秒 （-1 默认）
     *
     * @see RequestConfig#getConnectionRequestTimeout()
     */
    @Value("${spring.elasticsearch.socket-timeout:60000}")
    private Integer socketTimeout;

    /**
     * 连接请求超时/毫秒 （-1 默认）
     *
     * @see RequestConfig#getConnectionRequestTimeout()
     */
    @Value("${spring.elasticsearch.connection-request-timeout:5000}")
    private Integer connectionRequestTimeout;

    /**
     * 整个客户端的最大连接数
     */
    @Value("${spring.elasticsearch.max-conn-total:100}")
    private Integer maxConnTotal;

    /**
     * 每个节点的最大连接数，该数值应小于等于整个客户端连接数，如果时集群，则小于等于 最大连接数/节点数
     */
    @Value("${spring.elasticsearch.max-conn-per-route:20}")
    private Integer maxConnPerRoute;

    @Bean
    @ConditionalOnMissingBean(ElasticsearchClient.class)
    public ElasticsearchClient elasticsearchClient() {
        HttpHost[] httpHosts = toHttpHost();
        RestClientBuilder builder = RestClient.builder(httpHosts);

//        if (StringUtils.isNotBlank(userName)) {
        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(
                AuthScope.ANY, new UsernamePasswordCredentials(userName, passWord));

//            builder.setHttpClientConfigCallback(hc -> hc.setDefaultCredentialsProvider(credentialsProvider));
//        }

//        // 创建保持活动策略
//        ConnectionKeepAliveStrategy keepAliveStrategy = (response, context) -> {
//            // 从响应中获取 Keep-Alive 头部，如果不存在则使用默认的 5 秒
//            long keepAlive = 5 * 1000; // 5 秒
//            if (response.getHeaders("Keep-Alive").length > 0) {
//                // 解析响应头中的 Keep-Alive 值
//                // 这里可以添加具体解析逻辑
//            }
//            return keepAlive;
//        };

        // 配置 HTTP 客户端层级的设置
        builder.setHttpClientConfigCallback(httpClientBuilder -> {
            // 设置连接管理器、保持活动策略、默认连接配置和请求配置
            httpClientBuilder
                    .setDefaultCredentialsProvider(credentialsProvider)
//                    .setKeepAliveStrategy(keepAliveStrategy)
                    .setDefaultConnectionConfig(ConnectionConfig.custom().setCharset(StandardCharsets.UTF_8).build())   // 设置字符集
                    .setMaxConnTotal(maxConnTotal)// 整个客户端的最大连接数
                    .setMaxConnPerRoute(maxConnPerRoute);  // 每个节点的最大连接数，该数值应小于等于整个客户端连接数，如果时集群，则小于等于 最大连接数/节点数
            return httpClientBuilder;
        });

        // 配置请求级别的设置
        builder.setRequestConfigCallback(
                requestConfigBuilder -> requestConfigBuilder
                        .setConnectTimeout(connectTimeout) // 连接超时
                        .setSocketTimeout(socketTimeout) // Socket 超时
                        .setConnectionRequestTimeout(connectionRequestTimeout) // 请求超时
        );

        RestClient restClient = builder.build();
        // 创建ObjectMapper实例并注册JavaTimeModule
        ObjectMapper objectMapper = new ObjectMapper().registerModule(new JavaTimeModule());
        JacksonJsonpMapper jsonpMapper = new JacksonJsonpMapper(objectMapper);
        ElasticsearchTransport transport = new RestClientTransport(restClient, jsonpMapper);
        LOGGER.info("=====> Starting Elasticsearch");


        return new ElasticsearchClient(transport);
    }

    private HttpHost[] toHttpHost() {
        if (StringUtils.isBlank(hosts)) {
            throw new BusinessException("invalid elasticsearch configuration. elasticsearch.hosts不能为空！");
        }
        // 多个IP逗号隔开
        String[] hostArray = hosts.split(",");
        HttpHost[] httpHosts = new HttpHost[hostArray.length];
        HttpHost httpHost;
        for (int i = 0; i < hostArray.length; i++) {
            String[] strings = hostArray[i].split(":");
            httpHost = new HttpHost(strings[0], Integer.parseInt(strings[1]), "http");
            httpHosts[i] = httpHost;
        }
        return httpHosts;
    }

}
